package mr

import (
	"encoding/json"
	"fmt"
	"hash/fnv"
	"io/ioutil"
	"log"
	"net/rpc"
	"os"
	"sort"
	"strconv"
	"time"
)

//KeyValue :
// Map functions return a slice of KeyValue.
//
type KeyValue struct {
	Key   string
	Value string
}

// ByKey : for sorting by key.m
type ByKey []KeyValue

// for sorting by key.
func (a ByKey) Len() int           { return len(a) }
func (a ByKey) Swap(i, j int)      { a[i], a[j] = a[j], a[i] }
func (a ByKey) Less(i, j int) bool { return a[i].Key < a[j].Key }

//
// use ihash(key) % NReduce to choose the reduce
// task number for each KeyValue emitted by Map.
//
func ihash(key string) int {
	h := fnv.New32a()
	h.Write([]byte(key))
	return int(h.Sum32() & 0x7fffffff)
}

// Worker :
// main/mrworker.go calls this function.
//
func Worker(mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {

	// Your worker implementation here.

	// uncomment to send the Example RPC to the master.
	//CallExample()
	var taskid int
	var filename string
	var status int
	taskid, filename, status = AskTask()
	for true {
		switch status {
		case 0:
			//fmt.Printf("doWork: taskid %v, filename %v\n", taskid, filename)
			doWork(taskid, filename, true, mapf, reducef)
		case 1:
			doWork(taskid, filename, false, mapf, reducef)
		case 2:
			time.Sleep(time.Second)
		case 3:
			return
		default:
			return
		}
		taskid, filename, status = AskTask()
	}

}

func doWork(taskid int, filename string, ismap bool,
	mapf func(string, string) []KeyValue,
	reducef func(string, []string) string) {
	intermediate := []KeyValue{}
	if ismap {
		file, err := os.Open(filename)
		if err != nil {
			log.Fatalf("cannot open %v", filename)
		}
		content, err := ioutil.ReadAll(file)
		if err != nil {
			log.Fatalf("cannot read %v", filename)
		}
		file.Close()
		kva := mapf(filename, string(content))
		intermediate = append(intermediate, kva...)

		sort.Sort(ByKey(intermediate))

		intermediatefile := []*os.File{}
		encs := []*json.Encoder{}
		i := 0
		for i < 10 {
			oname := "mr-" + strconv.Itoa(taskid) + "-" + strconv.Itoa(i)
			ofile, err := os.Create(oname)
			if err != nil {
				log.Fatalf("cannot create %v", oname)
			}
			intermediatefile = append(intermediatefile, ofile)
			encs = append(encs, json.NewEncoder(ofile))
			i++
		}

		for _, kv := range intermediate {
			filenum := ihash(kv.Key) % 10
			err := encs[filenum].Encode(&kv)
			if err != nil {
				log.Fatalf("encode error %v", encs[filenum])
			}
		}
	} else {
		i := 0
		for i < 10 {
			redfilename := "mr-" + strconv.Itoa(i) + "-" + strconv.Itoa(taskid)
			redfile, _ := os.Open(redfilename)
			dec := json.NewDecoder(redfile)
			for {
				var kv KeyValue
				if err := dec.Decode(&kv); err != nil {
					break
				}
				intermediate = append(intermediate, kv)
			}
			redfile.Close()
			i++
		}

		sort.Sort(ByKey(intermediate))

		oname := "mr-output-" + strconv.Itoa(taskid)
		ofile, _ := os.Create(oname)
		i = 0
		for i < len(intermediate) {
			j := i + 1
			for j < len(intermediate) && intermediate[j].Key == intermediate[i].Key {
				j++
			}
			values := []string{}
			for k := i; k < j; k++ {
				values = append(values, intermediate[k].Value)
			}
			output := reducef(intermediate[i].Key, values)

			// this is the correct format for each line of Reduce output.
			fmt.Fprintf(ofile, "%v %v\n", intermediate[i].Key, output)

			i = j
		}
	}
	JobDone(taskid)
}

//CallExample :
// example function to show how to make an RPC call to the master.
//
// the RPC argument and reply types are defined in rpc.go.
//
func CallExample() {

	// declare an argument structure.
	args := ExampleArgs{}

	// fill in the argument(s).
	args.X = 99

	// declare a reply structure.
	reply := ExampleReply{}

	// send the RPC request, wait for the reply.
	call("Master.Example", &args, &reply)

	// reply.Y should be 100.
	fmt.Printf("reply.Y %v\n", reply.Y)
}

//AskTask :
func AskTask() (int, string, int) {
	args := JobDoneArgs{}
	reply := AsktaskReply{}
	if call("Master.Distributetask", &args, &reply) {
		return reply.Taskid, reply.Filename, reply.Status
	}
	return reply.Taskid, reply.Filename, 3
}

//JobDone : Tell master job done
func JobDone(taskid int) {
	args := JobDoneArgs{}
	args.IsDone = true
	args.Taskid = taskid
	reply := AsktaskReply{}
	call("Master.JobDone", &args, &reply)
}

//
// send an RPC request to the master, wait for the response.
// usually returns true.
// returns false if something goes wrong.
//
func call(rpcname string, args interface{}, reply interface{}) bool {
	// c, err := rpc.DialHTTP("tcp", "127.0.0.1"+":1234")
	sockname := masterSock()
	c, err := rpc.DialHTTP("unix", sockname)
	if err != nil {
		//log.Fatal("dialing:", err)
		return false
	}
	defer c.Close()

	err = c.Call(rpcname, args, reply)
	if err == nil {
		return true
	}

	fmt.Println(err)
	return false
}
